package com.eddy.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

public class NettyReactor {

    private ServerSocketChannel serverSocketChannel;

    private ReactorThread[] boss = new ReactorThread[1];

    private ReactorThread[] worker = new ReactorThread[2];

    public void newGroup() {

        for(int i=0; i<boss.length; i++) {
            boss[i] = new ReactorThread() {
                @Override
                void handle(SelectableChannel channel) {
                    AtomicInteger incr = new AtomicInteger(0);
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) channel;
                    try {
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        socketChannel.configureBlocking(false);
                        int index = incr.getAndIncrement() % worker.length;
                        ReactorThread workEventLoop = worker[index];
                        workEventLoop.doStart();
                        SelectionKey selectionKey = workEventLoop.register(socketChannel);
                        selectionKey.interestOps(SelectionKey.OP_READ);
                        System.out.println("接收到一个客户端连接：" + socketChannel.getRemoteAddress());
                    } catch (ClosedChannelException e) {
                        e.printStackTrace();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            };
        }

        for(int i=0; i<worker.length; i++) {
            worker[i] = new ReactorThread() {
                @Override
                void handle(SelectableChannel channel) {
                    SocketChannel socketChannel = (SocketChannel)channel;
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    try {
                        while(socketChannel.isOpen() && socketChannel.read(buffer) != -1) {
                            if(buffer.position() > 0) break;
                        }

                        if(buffer.position() == 0) return;

                        buffer.flip();
                        byte[] b = new byte[buffer.remaining()];
                        buffer.get(b);
                        System.out.println("收到客户端信息：" + new String(b, "utf-8"));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            };
        }
    }

    public void initAndRegister() throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        // 2、 将serverSocketChannel注册到selector
        int index = new Random().nextInt(boss.length);
        boss[index].doStart();
        SelectionKey selectionKey = boss[index].register(serverSocketChannel);
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
    }

    public void bind() throws IOException {
        serverSocketChannel.bind(new InetSocketAddress(9000));
        System.out.println("server is started...");
    }

    public static void main(String[] args) throws IOException {

        NettyReactor reactor = new NettyReactor();
        reactor.newGroup();
        reactor.initAndRegister();
        reactor.bind();
    }

    private abstract class ReactorThread extends Thread {

        Selector selector;

        public ReactorThread() {
            try {
                this.selector = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        public void doStart() {
            this.start();
        }

        @Override
        public void run() {
            try {
                while(true) {
                    selector.select(1000);

                    Set<SelectionKey> set = selector.selectedKeys();
                    Iterator<SelectionKey> itor = set.iterator();
                    while(itor.hasNext()) {
                        SelectionKey key = itor.next();
                        itor.remove();

                        if(key.isValid()) {
                            if (key.isReadable() || key.isAcceptable()) {
                                SelectableChannel selectableChannel = (SelectableChannel) key.attachment();
                                selectableChannel.configureBlocking(false);
                                handle(selectableChannel);
                            }
                        }

                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        public SelectionKey register(SelectableChannel channel) throws IOException {

            return channel.register(selector, 0, channel);
        }

        abstract void handle(SelectableChannel channel);
    }

}
